package com.raven.play.stream.test.javadsl.tutorial;

import akka.Done;
import akka.NotUsed;
import akka.japi.Pair;
import akka.stream.KillSwitches;
import akka.stream.SharedKillSwitch;
import akka.stream.javadsl.*;
import com.raven.play.stream.test.BaseTest;
import org.junit.jupiter.api.Test;

import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.Arrays;
import java.util.concurrent.CompletionStage;

public class $21DynamicStreamHandling extends BaseTest {

    @Test
    public void test1() {
        SharedKillSwitch sharedKillSwitch = KillSwitches.shared("oneButtonToRuleThemAll");

        Source<Integer, NotUsed> source1 = Source.range(1, 100).throttle(1, Duration.of(1, ChronoUnit.SECONDS)).log("source1");
        Source<Integer, NotUsed> source2 = Source.range(100, 200).throttle(1, Duration.of(2, ChronoUnit.SECONDS)).log("source2");
        Sink<Integer, CompletionStage<Done>> sink = Sink.foreach(it -> log.info(it.toString()));

        source1.via(sharedKillSwitch.flow()).to(sink).run(system);
        source2.via(sharedKillSwitch.flow()).to(sink).run(system);

        system.scheduler().scheduleOnce(Duration.of(3, ChronoUnit.SECONDS), () -> {
            log.info("Shutdowning");
            sharedKillSwitch.shutdown();
        }, system.executionContext());

    }

    /**
     * 动态合并, 有点像跑起来之后, 再将数据扔进对应的流中
     */
    @Test
    public void test2(){
        Source<String, Sink<String, NotUsed>> merge = MergeHub.of(String.class);
        Sink<String, Source<String, NotUsed>> broadcast = BroadcastHub.of(String.class);
        Pair<Sink<String, NotUsed>, Source<String, NotUsed>> run = merge.toMat(broadcast, Keep.both()).run(system);
        Sink<String, NotUsed> publisherPort = run.first();
        Source<String, NotUsed> subscriberPort = run.second();

        subscriberPort.runWith(Sink.foreach(e -> log.info("received: {}", e)), system);
        subscriberPort.map(String::length)
                .runWith(Sink.foreach(n -> System.out.println("I got a number: " + n)), system);

        Source.from(Arrays.asList("Akka", "is", "amazing")).runWith(publisherPort, system);
        Source.from(Arrays.asList("I", "love", "Scala")).runWith(publisherPort, system);
        Source.single("STREEEEEEAMS").runWith(publisherPort, system);
    }

}
